package com.nhncorp.pleiades.master;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClient;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.hadoop.New_RPC;
import com.nhncorp.pleiades.hadoop.New_Server;
import com.nhncorp.pleiades.protocol.RemoteAgentProtocol;
import com.nhncorp.pleiades.util.NetworkUtil;
import com.nhncorp.pleiades.util.Sleeper;

public class RemoteAgent implements RemoteAgentInterface{

  private LockService lockService;
  private New_Server server;  
  protected final AtomicBoolean stopRequested = new AtomicBoolean(false);
  private final Sleeper sleeper;
  public static final Log LOG = LogFactory.getLog(RemoteAgent.class
      .getName());
  
  public RemoteAgent(PleiadesUser userInfo) {
    
    PleiadesConfiguration conf = new PleiadesConfiguration();
    sleeper = new Sleeper(1000, stopRequested);
    InetSocketAddress serverAddress;
    try {
      serverAddress = NetworkUtil.getAddress(InetAddress
          .getLocalHost().getHostName()
          + ":" + conf.getInt("remoteAgent.port", 11000));
      this.server = New_RPC.getServer(this, serverAddress.getHostName(),
          serverAddress.getPort(), conf
              .getInt("remoteAgent.handler.count", 3), false, conf);
      lockService = new LockServiceClient(userInfo, conf);
      this.server.start();
    } catch (UnknownHostException e) {
      e.printStackTrace();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
  
  public String dumpLock() {
    LOG.info("RemoteAgent dumpLock ");
    try {
      return lockService.getAllLockStatus();
    } catch (IOException e) {
      LOG.error("dumpLock error", e);
      return null;
    }
  }

  public void lock(String path) {
    LOG.info("RemoteAgent Lock " + path);
    try {
      lockService.getLock(path, LockService.LockMode.LOCK_EXCLUSIVE);
    } catch (IOException e) {
      LOG.error("lock error:" + path, e);
    }
  }

  public void release(String path) {
    LOG.info("RemoteAgent Release " + path);
    try {
      lockService.releaseLock(path, LockService.LockMode.LOCK_EXCLUSIVE);
    } catch (IOException e) {
      LOG.error("releaseLock error:" + path, e);
    }
  }
  
  public long getProtocolVersion(String protocol, long clientVersion)
      throws IOException {
    if (protocol.equals(RemoteAgentProtocol.class.getName())) {
      return RemoteAgentProtocol.versionID;
    } else {
      throw new IOException("Unknown protocol to remote agent: " + protocol);
    }
  }
  
  public void shutdown() {
    stopRequested.set(true);
    lockService.close();
    server.stop();
  }
  
  public void run() {
    try {
      LOG.info("Remote Agent Thread Started");
      this.server.join();
    } catch (Throwable e) {
      if (e instanceof InterruptedException) {
        return;
      } else {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) throws Exception {

    RemoteAgent agent = null;
    switch(args.length) {
    case(0):
      agent= new RemoteAgent(new PleiadesUser("default"));
      break;
    case(1):
      agent= new RemoteAgent(new PleiadesUser(args[0]));
      break;
    case(2):
      agent= new RemoteAgent(new PleiadesUser(args[0], args[1]));
      break;
    default:
      break;
    }
    new Thread(agent, "Remote Agent Thread").start();
  }


}
